package com.ywu.mq.sdk.producer;

import com.ywu.boot.context.YwuContext;
import com.ywu.boot.context.YwuContextHolder;
import com.ywu.common.gson.GsonUtils;
import com.ywu.common.result.CommonResult;
import com.ywu.common.utils.TimerIdGenerateUtil;
import com.ywu.mq.sdk.config.YwuKafkaSdkProduceConfig;
import com.ywu.mq.sdk.domain.entity.YwuKafkaProduceMessageEntity;
import com.ywu.mq.sdk.utils.HttpUtil;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.*;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName YwuKafkaProducer
 * @Description kafka生产者模板方法
 * @Author GroundDemo
 * @Date 2025/3/28 9:51
 * @Version 1.0
 **/
@Component
@Slf4j
public abstract class YwuKafkaProducer {
    private YwuKafkaSdkProduceConfig sdkConfig;

    protected static final Properties PROPS = new Properties();

    private static final String PRODUCE_CHECK_TOPIC_URL = "/ywu/mq/kafka/sdk/kafka.produce.check.topic.info";

    private static final String PRODUCE_CREATE_MESSAGE_URL = "/ywu/mq/kafka/sdk/kafka.create.produce.message.info";

    private OkHttpClient httpClient;

    public void init(YwuKafkaSdkProduceConfig config) {
        // 连接池配置：最大空闲连接数10，连接保持存活时间5分钟
        ConnectionPool connectionPool = new ConnectionPool(10, 5, TimeUnit.MINUTES);
        // 构建http对象
        httpClient = new OkHttpClient.Builder()
                // 连接超时（建立连接的超时时间）
                .connectTimeout(config.getConnectTimeout(), TimeUnit.SECONDS)
                // 读取超时（从服务器获取响应数据的超时时间）
                .readTimeout(config.getReadTimeout(), TimeUnit.SECONDS)
                // 写入超时（向服务器发送数据的超时时间）
                .writeTimeout(config.getWriteTimeout(), TimeUnit.SECONDS)
                // 允许重定向
                .followRedirects(true)
                // 配置连接池
                .connectionPool(connectionPool)
                .build();
        this.sdkConfig = config;
        PROPS.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapService());
        PROPS.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer());
        PROPS.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,config.getValueSerializer());
        // 初始化权限校验
        if (sdkConfig.getEnablePermissionCheckBegin()) {
            // 调用spi查询主题是否存在并进行权限验证
            YwuKafkaProduceMessageEntity ywuKafkaProduceMessageEntity = new YwuKafkaProduceMessageEntity();
            ywuKafkaProduceMessageEntity.setAppId(sdkConfig.getAppId());
            ywuKafkaProduceMessageEntity.setTopicAppId(sdkConfig.getTopicAppId());
            ywuKafkaProduceMessageEntity.setAppKey(sdkConfig.getAppKey());
            ywuKafkaProduceMessageEntity.setTopicName(sdkConfig.getTopicName());
            // 构建请求体
            String serialize = GsonUtils.serialize(ywuKafkaProduceMessageEntity);
            // 构建URL
            String url = sdkConfig.getMqPlatformHost() + PRODUCE_CHECK_TOPIC_URL;
            String requestRes = HttpUtil.doPost(url, serialize, httpClient);
            CommonResult checkRes = GsonUtils.deserialize(requestRes, CommonResult.class);
            if (!"200".equals(checkRes.getCode())) {
                log.error("check begin topic is not exit OR topic permission is refused");
                throw new RuntimeException("topic is not exit OR topic permission is refused");
            }
        }
    }

    /**
    * @Author GroundDemo
    * @Description kafka生产的key
    * @Date 10:00 2025/3/28
    * @Param []
    * @return java.lang.String
    **/
    public String getKey() {
        return "";
    }

    /**
     * @Author GroundDemo
     * @Description kafka生产的topic
     * @Date 10:00 2025/3/28
     * @Param []
     * @return java.lang.String
     **/
    public String getTopic() {
        return "";
    }

    /**
    * @Author GroundDemo
    * @Description 生产消息的应用ID
    * @Date 12:29 2025/3/29
    * @Param []
    * @return java.lang.String
    **/
    public String getAppId() {
        return "";
    }

    /**
     * @Author GroundDemo
     * @Description 主题所属的应用ID
     * @Date 12:29 2025/3/29
     * @Param []
     * @return java.lang.String
     **/
    public String getTopicAppId() {
        return "";
    }

    /**
    * @Author GroundDemo
    * @Description 自定义生产回调
    * @Date 10:51 2025/3/28
    * @Param [metadata, exception, messageId]
    * @return void
    **/
    public abstract void customCallBack(RecordMetadata metadata, Exception exception, String messageId, String key, String value);

    /**
     * 发送后的回调
     *
     * @param metadata metadata
     * @param exception exception
     * @param messageEntity messageEntity
     * @param currentContext currentContext
     * @param record record
     */
    private void doProduceCallBack(RecordMetadata metadata, Exception exception, YwuKafkaProduceMessageEntity messageEntity,
                                   YwuContext currentContext, ProducerRecord<String, String> record) {
        YwuContextHolder.setCurrentContext(currentContext);
        if (exception != null) {
            log.error("produce error {}", exception.getMessage());
            messageEntity.setStatus(YwuKafkaProduceMessageEntity.PRODUCE_ERROR);
        } else {
            messageEntity.setStatus(YwuKafkaProduceMessageEntity.PRODUCE_SUCCESS);
            log.debug("produce send success partition {}, offset {}", metadata.partition(), metadata.offset());
        }
        // 调用spi记录生产日志
        // 构建请求体
        String serialize = GsonUtils.serialize(messageEntity);
        // 构建URL
        String url = sdkConfig.getMqPlatformHost() + PRODUCE_CREATE_MESSAGE_URL;
        HttpUtil.doPost(url, serialize, httpClient);
        // 执行自定义回调
        customCallBack(metadata, exception, messageEntity.getId(), record.key(), record.value());
    }

    /**
    * @Author GroundDemo
    * @Description 生产消息
    * @Date 9:57 2025/3/28
    * @Param []
    * @return void
    **/
    public void produce(String value) {
        YwuContext currentContext = YwuContextHolder.getCurrentContext();
        // 创建生产者实例
        try (Producer<String, String> producer = new KafkaProducer<>(PROPS)) {
            // 生产的key
            String key = getKey();
            if (StringUtils.isEmpty(key)) {
                key = sdkConfig.getKey();
            }
            // 主题所属的应用ID
            String topicAppId = getTopicAppId();
            if (StringUtils.isEmpty(topicAppId)) {
                topicAppId = sdkConfig.getTopicAppId();
            }
            // 生产这个消息的应用ID
            String appId = getAppId();
            if (StringUtils.isEmpty(appId)) {
                appId = sdkConfig.getAppId();
            }
            // 生产这个消息的appKey
            String appKey = sdkConfig.getAppKey();
            if (StringUtils.isEmpty(appKey)) {
                appKey = sdkConfig.getAppKey();
            }
            // 要生产到的主题
            String topic = getTopic();
            if (StringUtils.isEmpty(topic)) {
                topic = sdkConfig.getTopicName();
            }
            // 主题名称、主题所属的应用ID、生产这个消息的应用ID以及生产的key都不能为空，否则报错
            if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(topicAppId) || StringUtils.isEmpty(appId)
                    || StringUtils.isEmpty(key) || StringUtils.isEmpty(appKey)) {
                throw new RuntimeException("topic, topicAppId, appId, key, appKey can not be empty");
            }
            String replaceAppId = topicAppId.replace(".", "_").replace("-", "_");
            String topicName = replaceAppId + "-" + topic;
            // 构建对象
            YwuKafkaProduceMessageEntity ywuKafkaProduceMessageEntity = new YwuKafkaProduceMessageEntity();
            ywuKafkaProduceMessageEntity.setTopicName(topic);
            ywuKafkaProduceMessageEntity.setId(TimerIdGenerateUtil.nextId());
            ywuKafkaProduceMessageEntity.setKey(key);
            ywuKafkaProduceMessageEntity.setValue(value);
            ywuKafkaProduceMessageEntity.setStatus(YwuKafkaProduceMessageEntity.PRODUCING);
            ywuKafkaProduceMessageEntity.setAppId(appId);
            ywuKafkaProduceMessageEntity.setTopicAppId(topicAppId);
            ywuKafkaProduceMessageEntity.setAppKey(appKey);
            // 非初始化校验，每次生产消息的时候都需要校验
            if (!sdkConfig.getEnablePermissionCheckBegin()) {
                // 构建请求体
                String serialize = GsonUtils.serialize(ywuKafkaProduceMessageEntity);
                // 构建请求URL
                String url = sdkConfig.getMqPlatformHost() + PRODUCE_CHECK_TOPIC_URL;
                String requestRes = HttpUtil.doPost(url, serialize, httpClient);
                CommonResult checkRes = GsonUtils.deserialize(requestRes, CommonResult.class);
                if (!"200".equals(checkRes.getCode())) {
                    log.error("topic is not exit OR topic permission is refused");
                    throw new RuntimeException("topic is not exit OR topic permission is refused");
                }
            }
            // 构建新的value值，将消息ID存入value中 - 后续消费的时候会将其进行切割
            value = ywuKafkaProduceMessageEntity.getId() + "-ywuMessageId-" + value;
            // 构建ProducerRecord对象
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, value);
            producer.send(record, (metadata, exception) -> {
                // 生产回调
                doProduceCallBack(metadata, exception, ywuKafkaProduceMessageEntity, currentContext, record);
            });
        }
    }
}
